package com.diy.sigmund.diyorder.common.middleware.mq.rocketmq.spring.one;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;

/**
 * @author ylm-sigmund
 * @since 2022/5/5 22:19
 */
@RocketMQTransactionListener
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private static final Logger log = LoggerFactory.getLogger(TransactionListenerImpl.class);

    /**
     * <br>1.生产者向MQServer发送半消息【半消息：会存储进MQ Server，但是被标记为不能投递状态】
     * <br>2.发送半消息成功，生产者实行本地事务
     * <br>3.根据本地事务结果向MQ Server发送二次确认请求
     * <br>4.MQ Server根据接受到的消息投递或者丢弃消息
     * <br>5.若在本地事务执行过程中缺少二次确认消息或生产者处于等待状态，MQ服务器将向同一组中的每个生产者发送检查消息，然后继续3，4的操作
     * <br>
     * <br>我们可以利用 mq 的两个特性 重试 和 死信队列 来协助消费端处理：
     * <br>消费失败后进行一定次数的 重试
     * <br>重试后也失败的话该消息丢进 死信队列 里
     * <br>另外起一个线程监听消费 死信队列 里的消息，记录日志并且预警！
     * <br>因为有 重试 所以消费者需要实现 幂等性
     * <br>
     * <br>消息三态
     * <br>
     * <br>Commit：提交事务信息，消费者可以消费此消息
     * <br>Rollback：回滚事务消息，broker会删除这条消息，消费者不能消费
     * <br>UNKNOWN：broker需要回查确认消息状态
     */
    @SuppressWarnings("rawtypes")
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
            Object arg) {

        if ("1".equals(arg)) {
            log.info(
                    "executeLocalTransaction: " + new String((byte[]) msg.getPayload()) + " unknown");
            // 返回UNKNOWN时，会调用checkLocalTransaction()
            return RocketMQLocalTransactionState.UNKNOWN;
        } else if ("2".equals(arg)) {
            log.info(
                    "executeLocalTransaction: " + new String((byte[]) msg.getPayload()) + " rollback");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        log.info(
                "executeLocalTransaction: " + new String((byte[]) msg.getPayload()) + " commit");
        return RocketMQLocalTransactionState.COMMIT;
    }

    @SuppressWarnings("rawtypes")
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 从数据库查查询t_message_transaction表，如果该表中存在记录，则提交，
        // 然后t_message_transaction 表，是否存在bizUniNo，如果存在，则返回COMMIT_MESSAGE，
        // 不存在，则记录查询次数，未超过次数，返回UNKNOW，超过次数，返回ROLLBACK_MESSAGE
        log.info("checkLocalTransaction: " + new String((byte[]) msg.getPayload()));
        return RocketMQLocalTransactionState.COMMIT;
    }
}
